/**
 * Copyright © 2017 sdjictec.Co.Ltd. All rights reserved.
 **/
package com.sdjictec.haier;

import com.alibaba.fastjson.JSON;
import com.sdjictec.haier.config.ProFileConfig;
import com.sdjictec.haier.entity.RetailDataMQ;
import com.sdjictec.haier.utils.DBUtills;
import com.sdjictec.haier.utils.IJMSInfo;
import com.sdjictec.haier.utils.ILogic;
import org.apache.commons.cli.*;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.*;


public class Kafka2Oracletest {

    private static Logger logger = LoggerFactory.getLogger(Kafka2Oracletest.class);


    public static void main(String[] args) {

        CommandLine cmd = null;
        Options options = new Options();
        try {
//            Option source_tablenameO = new Option("test", true, "");
//            source_tablenameO.setRequired(true);
//            options.addOption(source_tablenameO);

            Option resetO = new Option("reset", true, "");
            resetO.setRequired(true);
            options.addOption(resetO);

            DefaultParser parser = new DefaultParser();
            cmd = parser.parse(options, args);
        } catch (ParseException e) {
            e.printStackTrace();
            logger.error(e.getMessage());

        }
//        String test = cmd.getOptionValue("test");
        String reset = cmd.getOptionValue("reset");

        logger.info("reset:" + reset);

        System.out.println("reset:" + reset);

        IJMSInfo ijmsInfo = new IJMSInfo();

        ILogic iLogic = new ILogic();


        DBUtills dbUtills = new DBUtills();


        Properties props = new Properties();
        props.put("bootstrap.servers", ProFileConfig.kafka_servers);
        props.put("group.id", ProFileConfig.kafka_group_id);
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        if (reset.equals("earliest")) {
            props.put("auto.offset.reset", "earliest");
        } else {
            props.put("auto.offset.reset", "latest");
        }

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(ProFileConfig.kafka_topic));

        while (true) {

            logger.info("start read date");

            System.out.println("start read date");


            try {
                ConsumerRecords<String, String> records = consumer.poll(100);

                if (records != null && !records.isEmpty()) {


                    for (ConsumerRecord<String, String> record : records) {

                        String strval = record.value();

//
//                        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                        Date now = new Date();
//                        if (Integer.valueOf(formatter.format(now).substring(14, 16)) <= 5) {
//                            try {
//                                Thread.sleep(480);
//                            } catch (InterruptedException e) {
//                                e.printStackTrace();
//                            }
//                        }
//
//
//                        System.out.println(strval);
//
//
//                        Map<String, Object> mapTypes = JSON.parseObject(strval);
//                        String flag = (String) mapTypes.get("flag");
//                        //                System.out.println(str);
//                        //只取flag为dj的数据
//                        if (!flag.equals(ProFileConfig.realtimeFlagFilter)) {
//
//                            continue;
//                        }
//
//                        List<Map<String, Object>> billDatas = (List<Map<String, Object>>) mapTypes.get("billData");
//                        for (Map<String, Object> billData : billDatas) {
//
//
//                            if (!ijmsInfo.filterFlag(mapTypes, billData)) {
//
//                                continue;
//
//                            }
//
//                            //解析MQ Json数据
//                            System.out.println(strval);
//                            logger.debug(strval);
//                            RetailDataMQ retailDataMQ = ijmsInfo.parseJson(mapTypes, billData);
//
//                            RetailDataMQ fromRedis = iLogic.mergeDataInfo(retailDataMQ);
//
//                            //                        RetailDataMQ2 retailDataMQ2 = new RetailDataMQ2(fromRedis);
//
//                            dbUtills.jdbcsave(fromRedis, "mq");
//
//                            dbUtills.jdbcsave(fromRedis, "mq2");
//
//                        }

                    }

                    try {
                        //                    consumer.commitSync();
                        consumer.commitAsync();

                    } catch (CommitFailedException e) {
                        e.printStackTrace();
                        logger.error(e.getMessage());
                        logger.error("commit failed", e);
                    }

                } else {
                    System.out.println("no data");

                    logger.info("no data");
                }
            } catch (NumberFormatException e) {


                e.printStackTrace();
                logger.error(e.getMessage());

            }
        }

    }

}